package com.navinfo.platform.thousandkilo.utils;

import com.navinfo.platform.hbase.api.RowMapper;
import com.navinfo.platform.thousandkilo.dto.ThousandKiloData;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * 对象与hbase格式转换
 */
public class HbaseConverter implements RowMapper<ThousandKiloData> {
    private static final byte[] D = Bytes.toBytes("d");
    private static final byte[] UID = Bytes.toBytes("uid");
    private static final byte[] TID = Bytes.toBytes("tid");
    private static final byte[] MILEAGE_SECTION = Bytes.toBytes("mileageSection");
    private static final byte[] SECTION_START_FUEL = Bytes.toBytes("sectionStartFuel");
    private static final byte[] SECTION_END_FUEL = Bytes.toBytes("sectionEndFuel");
    private static final byte[] SECTION_FUEL = Bytes.toBytes("sectionFuel");
    private static final byte[] SECTION_START_MILEAGE = Bytes.toBytes("sectionStartMileage");
    private static final byte[] SECTION_END_MILEAGE = Bytes.toBytes("sectionEndMileage");
    private static final byte[] SECTION_MILEAGE = Bytes.toBytes("sectionMileage");
    private static final byte[] SECTION_UREA = Bytes.toBytes("sectionUrea");

    public static String getRowKey(Long tid, Integer mileageSection) {
        return new StringBuffer(String.valueOf(tid)).reverse().toString() + "|" + mileageSection;
    }

    public static Put toPut(ThousandKiloData m) {
        Put put = new Put(Bytes.toBytes(getRowKey(m.getTid(), m.getMileageSection())));
        /*--start_schema*/
        if(m.getUid()!=null){put.addColumn(D, UID, toBytes(m.getUid()));}
        if(m.getTid()!=null){put.addColumn(D, TID, toBytes(m.getTid()));}
        if(m.getMileageSection()!=null){put.addColumn(D, MILEAGE_SECTION, toBytes(m.getMileageSection()));}
        if(m.getSectionStartFuel()!=null){put.addColumn(D, SECTION_START_FUEL, toBytes(m.getSectionStartFuel()));}
        if(m.getSectionEndFuel()!=null){put.addColumn(D, SECTION_END_FUEL, toBytes(m.getSectionEndFuel()));}
        if(m.getSectionFuel()!=null){put.addColumn(D, SECTION_FUEL, toBytes(m.getSectionFuel()));}
        if(m.getSectionStartMileage()!=null){put.addColumn(D, SECTION_START_MILEAGE, toBytes(m.getSectionStartMileage()));}
        if(m.getSectionEndMileage()!=null){put.addColumn(D, SECTION_END_MILEAGE, toBytes(m.getSectionEndMileage()));}
        if(m.getSectionMileage()!=null){put.addColumn(D, SECTION_MILEAGE, toBytes(m.getSectionMileage()));}
        if(m.getSectionUrea()!=null){put.addColumn(D, SECTION_UREA, toBytes(m.getSectionUrea()));}
        /*--end_schema*/
        return put;
    }

    @Override
    public ThousandKiloData mapRow(Result result, int rowNum) throws Exception {
        ThousandKiloData data = new ThousandKiloData();
        data.setUid(toLong(result.getValue(D, UID)));
        data.setTid(toLong(result.getValue(D, TID)));
        data.setMileageSection(toInt(result.getValue(D, MILEAGE_SECTION)));
        data.setSectionStartFuel(toInt(result.getValue(D, SECTION_START_FUEL)));
        data.setSectionEndFuel(toInt(result.getValue(D, SECTION_END_FUEL)));
        data.setSectionFuel(toInt(result.getValue(D, SECTION_FUEL)));
        data.setSectionStartMileage(toInt(result.getValue(D, SECTION_START_MILEAGE)));
        data.setSectionEndMileage(toInt(result.getValue(D, SECTION_END_MILEAGE)));
        data.setSectionMileage(toInt(result.getValue(D, SECTION_MILEAGE)));
        data.setSectionUrea(toInt(result.getValue(D, SECTION_UREA)));
        return data;
    }

    private static byte[] toBytes(int val) {
        return Bytes.toBytes(String.valueOf(val));
    }

    private static byte[] toBytes(long val) {
        return Bytes.toBytes(String.valueOf(val));
    }

    private static long toLong(byte[] bytes) {
        return Long.parseLong(Bytes.toString(bytes));
    }

    private static int toInt(byte[] bytes) {
        return Integer.parseInt(Bytes.toString(bytes));
    }
}
